package org.spring.aicloud.controller;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.xxl.job.core.handler.annotation.XxlJob;
import jakarta.annotation.Resource;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.redisson.api.RedissonClient;
import org.spring.aicloud.entity.Answer;
import org.spring.aicloud.util.AppVariable;
import org.spring.aicloud.util.idempotent.Idempotent;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.data.annotation.Reference;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * kafka 调用测试控制器
 */
@RefreshScope
@RestController
@RequestMapping("/kafka")
public class KafkaController {

    @Resource
    private KafkaTemplate kafkaTemplate;
    private static final String TEST_TOPIC = "ai-cloud-test-topic";
    // canal 将 mysql binlog 同步到 kafka 中的 topic
    private static final String CANAL_TOPIC = "ai-cloud-canal-to-kafka";
    @Resource
    private RedissonClient redissonClient;
    @Resource
    private ObjectMapper objectMapper;
    @Resource
    private RedisTemplate<String, Object> redisTemplate;


    @RequestMapping("/testlock")
    public String testLock() {
        String msg = "";
        String lockKey = AppVariable.getModelLockKey(1L, 1, 1);
        boolean isLock = false;
        try {
            isLock = redissonClient.getLock(lockKey).tryLock(10, TimeUnit.SECONDS);
            msg = "获得分布式锁：" + isLock;
            System.out.println(msg);
        } catch (Exception e) {
        } finally {
            redissonClient.getLock(lockKey).unlock();
        }
        if (!isLock) {
            msg = "获得分布式锁失败~";
            System.out.println(msg);
        }
        return msg;
    }

    @XxlJob("testJob")
    public void testJob() {
        System.out.println("执行了定时任务");
    }

    @RequestMapping("/send")
    public String send(String msg) {
        if (!StringUtils.hasLength(msg)) {
            return "请先输入发送的消息！";
        }
        kafkaTemplate.send(TEST_TOPIC, msg);
        return "ok";
    }

    @KafkaListener(topics = {TEST_TOPIC})
    public void listen(String msg, Acknowledgment acknowledgment) {
        System.out.println("接收到消息：" + msg);

        // 手动确认应答
        acknowledgment.acknowledge();
    }

    @KafkaListener(topics = {CANAL_TOPIC})
    public void canalListen(String data, Acknowledgment acknowledgment) throws JsonProcessingException {
        HashMap<String, Object> map = objectMapper.readValue(data, HashMap.class);
        if (!map.isEmpty() && map.get("database").toString().equals("aicloud") &&
                map.get("table").toString().equals("answer")) {
            // 更新 Redis 缓存
            ArrayList<LinkedHashMap<String, Object>> list =
                    (ArrayList<LinkedHashMap<String, Object>>) map.get("data");
            String cacheKey = "";
            for (LinkedHashMap<String, Object> answer : list) {
                cacheKey = AppVariable.getListCacheKey(
                        Long.parseLong(answer.get("uid").toString()),
                        Integer.parseInt(answer.get("model").toString()),
                        Integer.parseInt(answer.get("type").toString()));
                redisTemplate.opsForValue().set(cacheKey, null);
            }
        }
        // 手动确认应答
        acknowledgment.acknowledge();
    }

}
